package com.springboot.demo.config;

import cn.hutool.core.collection.ConcurrentHashSet;
import com.springboot.demo.entity.User;
import com.springboot.demo.entity.UserDetailsImpl;
import com.springboot.demo.entity.vo.WebsocketUserVO;
import com.springboot.demo.exception.BusinessException;
import com.springboot.demo.service.InboxService;
import com.springboot.demo.utils.Constant;
import com.springboot.demo.utils.RedisUtils;
import com.springboot.demo.utils.ResponseResult;
import com.sun.istack.Nullable;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.simp.SimpMessageType;
import org.springframework.messaging.simp.config.ChannelRegistration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
import org.springframework.web.socket.server.support.HttpSessionHandshakeInterceptor;

import java.security.Principal;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author wubc683
 * @desc
 * @date 2021-11-05 11:26
 */
@Configuration
@EnableWebSocketMessageBroker
@Slf4j
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
    /**
     * 这个Bean会自动注册使用@ServerEndpoint注解声明的websocket endpoint
     *
     * @param
     * @return: org.springframework.web.socket.server.standard.ServerEndpointExporter
     */
//    @Autowired
//    InboxService inboxService;
//    @Bean
//    public ServerEndpointExporter serverEndpointExporter() {
//        return new ServerEndpointExporter();
//    }
    //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
    private static int onlineCount = 0;
    // 静态变量，记录用户连接
    private static ConcurrentHashMap<String, Object> webSocketMap = new ConcurrentHashMap<>();

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        //如果不使用rabbitmq作为消息代理,那么只需要暴露简单节点即可
        // 默认的Simple Broker
//        registry.enableSimpleBroker("/topic", "/user");
//        registry.setApplicationDestinationPrefixes("/app");

        // 使用RabbitMQ做为消息代理，替换默认的Simple Broker
        //定义了客户端接收地址的前缀，也即服务端发送消息给客户端的地址前缀,@SendTo(XXX) 也可以重定向,订阅需要加上/user
        registry.setUserDestinationPrefix("/user");

        // 客户端推送消息到服务器时使用，客户端推送的接口加上/app,会路由到带有@MessageMapping注解的方法中
        registry.setApplicationDestinationPrefixes("/app");

        // "STOMP broker relay"处理所有消息将消息发送到外部的消息代理
        registry.enableStompBrokerRelay("/exchange", "/topic", "/queue", "/amq/queue")
                .setVirtualHost("/") //对应自己rabbitmq里的虚拟host
                //.setRelayHost("121.196.37.50")
                .setRelayHost("120.55.241.14")
                .setClientLogin("wubc683")
                .setClientPasscode("rbt1234*")
                .setSystemLogin("wubc683")
                .setSystemPasscode("rbt1234*")
                .setSystemHeartbeatSendInterval(10000)
                .setSystemHeartbeatReceiveInterval(10000);
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/stomp")
                .setAllowedOriginPatterns("*")
                .addInterceptors(new HttpSessionHandshakeInterceptor())
                .withSockJS();
    }

    @Override
    public boolean configureMessageConverters(List<MessageConverter> messageConverters) {
        return true;
    }

    @Override
    public void configureClientInboundChannel(ChannelRegistration registration) {
        ChannelInterceptor interceptor = new ChannelInterceptor() {
            @Override
            public Message<?> preSend(Message<?> message, MessageChannel channel) {
                StompHeaderAccessor accessor = StompHeaderAccessor.wrap(message);
                MessageHeaders header = message.getHeaders();
                String sessionId = (String) header.get("simpSessionId");
                UsernamePasswordAuthenticationToken simpUser = (UsernamePasswordAuthenticationToken)
                        accessor.getMessageHeaders().get("simpUser");
                UserDetailsImpl userDetailsImpl = (UserDetailsImpl) simpUser.getPrincipal();
                User user = userDetailsImpl.getUser();
                String subscribeUrl = (String) header.get("simpDestination");
                if (accessor.getCommand() != null) {
                    SimpMessageType type = accessor.getCommand().getMessageType();
                    if (SimpMessageType.CONNECT.equals(type)) {
                        // 新连接
                        if (webSocketMap.get(user.getUsername()) == null) {
                            webSocketMap.put(user.getUsername(), user);
                            onlineCount++;
                            log.info("用户{}加入了websocket连接,sessionId:{},websocket连接数：{}", user.getId(), sessionId, onlineCount);
                        } else {
                            throw new BusinessException(ResponseResult.TWO_MANY_CONNECTION_FOR_ONE_USER_ERROR);
                        }
                    } else if (type == SimpMessageType.DISCONNECT) {
                        if (webSocketMap.get(user.getUsername()) != null) {
                            webSocketMap.remove(user.getUsername());
                            onlineCount--;
                            log.info("用户{}离开了websocket连接,sessionId:{},websocket连接数：{}", user.getUsername(), sessionId, onlineCount);
                        } else {
                            return null;
                        }
                    } else if (type == SimpMessageType.SUBSCRIBE) {
                        if (subscribeUrl == null || (!"/topic/topic.web.public.routingkey".equals(subscribeUrl)
                                && !"/user/topic/topic.web.private.routingkey".equals(subscribeUrl))) {
                            log.warn("用户{}订阅url不合法,sessionId:{},url:{}", user.getUsername(), sessionId, subscribeUrl);
                            return null;
                        }
                        log.info("用户{}订阅了url；{},sessionId:{}", user.getUsername(), subscribeUrl, sessionId);
                    } else if (type == SimpMessageType.MESSAGE) {
                        log.info("消息url:{},user:{},sessionId:{},content:{}", subscribeUrl, user.getUsername(), sessionId,
                                message);
                    } else if (type == SimpMessageType.OTHER) {
                        log.info("other类型命令，消息url:{},user:{},sessionId:{},content:{}", subscribeUrl, user.getUsername(),
                                sessionId, message);
                    }
                }
                return message;
            }

            @Override
            public boolean preReceive(MessageChannel channel) {
                log.info("Inbound preReceive. channel={}", channel);
                return true;
            }

            @Override
            public Message<?> postReceive(Message<?> message, MessageChannel channel) {
                log.info("Inbound postReceive. message={}", message);
                return message;
            }

            @Override
            public void postSend(Message<?> message, MessageChannel channel, boolean sent) {
//                log.info("Inbound postSend. message={}", message);
            }

//            @Override
//            public void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, @Nullable Exception ex) {
//                log.info("Inbound afterSendCompletion. message={}", message);
//                StompHeaderAccessor accessor = StompHeaderAccessor.wrap(message);
//                MessageHeaders header = message.getHeaders();
//
//                UsernamePasswordAuthenticationToken simpUser = (UsernamePasswordAuthenticationToken)
//                        accessor.getMessageHeaders().get("simpUser");
//                UserDetailsImpl userDetailsImpl = (UserDetailsImpl) simpUser.getPrincipal();
//                User user = userDetailsImpl.getUser();
//
//                if (accessor.getCommand() != null) {
//                    SimpMessageType type = accessor.getCommand().getMessageType();
//                    if (type == SimpMessageType.SUBSCRIBE) {
//                        // 发送未确认消息给网页客户端
////                        inboxService.sendUnconfirmedMsg(user);
//                    }
//                }
//            }

            @Override
            public void afterReceiveCompletion(@Nullable Message<?> message, MessageChannel channel, @Nullable Exception ex) {
                log.info("Inbound afterReceiveCompletion. message={}", message);
            }
        };
        registration.interceptors(interceptor);
    }

}
